package org.sean.framework.kafka.logging.appender;

import ch.qos.logback.core.UnsynchronizedAppenderBase;
import ch.qos.logback.core.spi.AppenderAttachable;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.sean.framework.code.StatusInfo;
import org.sean.framework.exception.StatusException;
import org.sean.framework.kafka.logging.LogConstants;
import org.sean.framework.kafka.logging.delivery.DeliveryStrategy;
import org.sean.framework.kafka.logging.encoding.KafkaMessageEncoder;
import org.sean.framework.kafka.logging.keying.KeyingStrategy;
import org.sean.framework.util.StringUtil;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.LoggingProducerListener;
import org.springframework.lang.Nullable;

import java.util.ArrayList;
import java.util.Collections;

/**
 * 配置项
 *
 * @param <E> E
 * @author xielei
 */
public abstract class AbstractKafkaAppender<E> extends UnsynchronizedAppenderBase<E> implements AppenderAttachable<E> {

    protected String topic = null;
    protected String servers = null;
    protected String clientId = null;
    protected KafkaMessageEncoder<E> encoder = null;
    protected KeyingStrategy<? super E> keyingStrategy = null;
    protected DeliveryStrategy deliveryStrategy;

    public void setEncoder(KafkaMessageEncoder<E> layout) {
        this.encoder = layout;
    }

    public void setTopic(String topic) {
        if (topic.contains(LogConstants.KAFKA_TOPIC_PROPERTY)) {
            throw new StatusException(StatusInfo.serverInitError().setAnyMessage("Kafka topic must config"));
        }
        this.topic = topic;
    }

    public void setServers(String servers) {
        if (servers.contains(LogConstants.KAFKA_SERVERS_PROPERTY)) {
            throw new StatusException(StatusInfo.serverInitError().setAnyMessage("Kafka servers must config"));
        }
        this.servers = servers;
    }

    public void setClientId(String clientId) {
        this.clientId = clientId;
    }

    public void setKeyingStrategy(KeyingStrategy<? super E> keyingStrategy) {
        this.keyingStrategy = keyingStrategy;
    }

    public void setDeliveryStrategy(DeliveryStrategy deliveryStrategy) {
        this.deliveryStrategy = deliveryStrategy;
    }

    protected KafkaTemplate<Object, Object> kafkaTemplate() {
        if (StringUtil.isEmpty(topic) || StringUtil.isEmpty(servers)) {
            return null;
        }
        KafkaTemplate<Object, Object> kafkaTemplate = new KafkaTemplate<>(kafkaProducerFactory());
        kafkaTemplate.setDefaultTopic(topic);
        kafkaTemplate.setProducerListener(new LoggingProducerListener<Object, Object>() {
            @Override
            public void onError(ProducerRecord<Object, Object> record, @Nullable RecordMetadata recordMetadata, Exception exception) {
                // do nothing
            }
        });
        return kafkaTemplate;
    }

    private ProducerFactory<Object, Object> kafkaProducerFactory() {
        KafkaProperties properties = new KafkaProperties();
        properties.setBootstrapServers(new ArrayList<>(Collections.singletonList(servers)));
        properties.setClientId(clientId);
        DefaultKafkaProducerFactory<Object, Object> factory = new DefaultKafkaProducerFactory<>(properties.buildProducerProperties());
        String transactionIdPrefix = properties.getProducer().getTransactionIdPrefix();
        if (transactionIdPrefix != null) {
            factory.setTransactionIdPrefix(transactionIdPrefix);
        }
        return factory;
    }

}
